package com.xrui.hbase.filter;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.xrui.hbase.DataRequest;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;

import java.io.IOException;


public final class RowPrefixFilter extends RowFilter {

    /**
     * All StripValueRowFilter instances are the same: generate hash-code ahead of time.
     */
    private static final int HASH_CODE =
        new HashCodeBuilder().append(RowPrefixFilter.class).toHashCode();
    private byte[] mPrefix;

    public RowPrefixFilter(byte[] rowKeyPrefix) {
        Preconditions.checkState(rowKeyPrefix != null, "RowPrefixFilter should have a prefix.");
        mPrefix = rowKeyPrefix;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public DataRequest getDataRequest() {
        return DataRequest.builder().build();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter() throws IOException {
        return new PrefixFilter(mPrefix);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof RowPrefixFilter)) {
            return false;
        } else {
            final RowPrefixFilter otherFilter = (RowPrefixFilter) other;
            return Objects.equal(otherFilter.mPrefix, mPrefix);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return HASH_CODE;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(RowPrefixFilter.class).toString();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected JsonNode toJsonNode() {
        final ObjectNode root = JsonNodeFactory.instance.objectNode();
        root.put("prefix", mPrefix);
        return root;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected Class<? extends RowPrefixFilterDeserializer> getDeserializerClass() {
        return RowPrefixFilterDeserializer.class;
    }

    /**
     * Deserializes {@code HasColumnDataRowFilter}.
     */
    public static final class RowPrefixFilterDeserializer
        implements RowFilterDeserializer {
        /**
         * {@inheritDoc}
         */
        @Override
        public RowFilter createFromJson(JsonNode root) {
            final String prefix = root.path("prefix").getTextValue();
            return new RowPrefixFilter(prefix.getBytes());
        }
    }

}
